package com.fzk.rocketmq;

import com.fzk.log.Logger;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import java.io.IOException;
import java.util.Map;

/**
 * @author zhike.feng
 * @datetime 2023-11-26 23:13:01
 */
public class ConsumerDemo implements AutoCloseable {
    public static void main(String[] args) {
        try (ConsumerDemo consumerDemo = new ConsumerDemo()) {
            Thread.sleep(1000 * 60);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private final RocketMQConf conf;
    private final PushConsumer consumer;
    private final ClientServiceProvider provider;

    private ConsumerDemo() throws ClientException {
        this.provider = ClientServiceProvider.loadService();
        this.conf = RocketMQConf.getConf();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(conf.getEndpoint())
                .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        // 为消费者指定所属的消费者分组，Group需要提前创建。
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。


        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        this.consumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // 设置消费者分组。
                .setConsumerGroup(conf.getConsumerGroup())
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(Map.of(conf.getTopic(), filterExpression))
                // 设置消费监听器。
                .setMessageListener(msg -> {
                    //  处理消息并返回消费结果。
                    byte[] a = new byte[msg.getBody().remaining()];
                    msg.getBody().get(0, a);
                    Logger.info(String.format("Consume message successfully, messageInfo: %s, \tcontent: %s", msg.toString(), new String(a)));
                    return ConsumeResult.SUCCESS;
                })
//                .setAwaitDuration(Duration.ofSeconds(10)) // 设置长轮询等待时间
                .build();
//        this.consumer.receive(10, Duration.ofSeconds(10)) // 设置消息对同组内其它消费者不可见时间 -> 其实就是业务处理超时时间
//                .forEach(msg -> {
//                    // 处理消息并返回消费结果。
//                    byte[] a = new byte[msg.getBody().remaining()];
//                    msg.getBody().get(0, a);
//                    Logger.info(String.format("Consume message successfully, messageInfo: %s, \tcontent: %s", msg.toString(), new String(a)));
//                    try {
//                        this.consumer.ack(msg);
//                    } catch (ClientException e) {
//                        throw new RuntimeException(e);
//                    }
//                });
    }


    @Override
    public void close() {
        try {
            this.consumer.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
